package org.ala.tiktools.rpc.kcp.protocol.sender;

import org.ala.tiktools.rpc.kcp.config.RpcKcpConfigInfo;
import org.ala.tiktools.rpc.kcp.protocol.IKcp;
import org.ala.tiktools.rpc.kcp.protocol.IKcpOutput;
import org.ala.tiktools.rpc.kcp.protocol.Kcp;
import org.ala.tiktools.rpc.kcp.protocol.receiver.AckList;
import org.ala.tiktools.rpc.kcp.protocol.segment.KcpSegment;
import org.ala.tiktools.rpc.kcp.protocol.sender.queue.KcpSenderLinkedList;
import org.ala.tiktools.tools.list.DoubleLinkedList;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;

/**
 * @author ala
 * @date 2025-04-11 23:22
 */
public class KcpSender {
    /**
     *  绑定的kcp
     */
    protected Kcp kcp;


    /**
     *  下一个要接收ack的sn
     *  <p>	从1开始，0空出来判断绕回
     */
    protected long sndUna = 1;
    /**
     *  下一个要发送的sn
     *  <p>	从1开始，0空出来判断绕回
     */
    protected long sndNxt = 1;
    /**
     *  发送窗口大小
     */
    protected int sndWnd;
    /**
     *  发送缓冲
     */
    protected KcpSenderLinkedList sndBuf;
    /**
     *  待发送队列
     */
    protected KcpSenderLinkedList sndQue;
    /**
     *  真实发送工具
     */
    protected IKcpOutput output;
    /**
     *
     */
    protected ByteBufAllocator byteBufAllocator = ByteBufAllocator.DEFAULT;


    public KcpSender(Kcp kcp, int sndWnd, IKcpOutput output) {
        this.kcp = kcp;
        this.sndWnd = sndWnd;
        sndBuf = new KcpSenderLinkedList();
        sndQue = new KcpSenderLinkedList();
        this.output = output;
    }


    /**
     *  是否可以发送数据
     *  @param  curCanSend  是否现在发送<br/>
     *                      true：sndBuf + sndQue < sndWnd*2，可发送
     *                      false：sndBuf + sndQue < sndWnd，可发送
     */
    public boolean canSend(boolean curCanSend) {
        int max = sndWnd;
        int waitSnd = sndBuf.size() + sndQue.size();
        //  curCanSend模式下，待发送数据不超过 sndWnd*2，就可以发送
        if (curCanSend) {
            return waitSnd < (max<<1);
        }
        //  否则待发送数据不能超过 sndWnd
        else {
            int threshold = Math.max(1, max);
            return waitSnd < threshold;
        }
    }
    /**
     *  待发送的数据总量
     */
    public int waitSnd() {
        return sndBuf.size() + sndQue.size();
    }


    /**
     *  写入ackList，完了清空ackList
     */
    public ByteBuf writeAndClearAckList(AckList ackList, ByteBuf buf, KcpSegment seg) {
        //  ackList是跟着channel事件被触发的，channel事件只会在与其绑定eventLoop中被触发，而一个channel只会绑定一个eventLoop线程
        for (int i = 0 ; i < ackList.size() ; i++) {
            buf = createOrOutput(buf, IKcp.IKCP_OVERHEAD);
            int p = i << 1;
            int sn = ackList.indexOf(p), ts = ackList.indexOf(p+1);
            seg.setCmd(IKcp.IKCP_CMD_ACK);
            seg.setSn(sn & 0xFFFFFFFFL);    //  转32位无符号
            seg.setTs(ts);
            seg.writeHead(buf);
        }
        //  清空ackList
        ackList.clear();
        return buf;
    }
    /**
     *  写入询问窗口大小
     */
    public ByteBuf writeAskWnd(ByteBuf buf, KcpSegment seg) {
        buf = createOrOutput(buf, IKcp.IKCP_OVERHEAD);
        seg.setCmd(IKcp.IKCP_CMD_WASK);
        seg.writeHead(buf);
        return buf;
    }
    /**
     *  写入回答窗口大小消息（当前窗口大小，已经在模版里给出）
     */
    public ByteBuf writeTellWnd(ByteBuf buf, KcpSegment seg) {
        buf = createOrOutput(buf, IKcp.IKCP_OVERHEAD);
        seg.setCmd(IKcp.IKCP_CMD_WINS);
        seg.writeHead(buf);
        return buf;
    }
    /**
     *  尝试创建或发送数据
     *  @param  buf 要发送数据的buf
     *  @param  len 要发送的数据长度
     */
    protected ByteBuf createOrOutput(ByteBuf buf, int len) {
        if (buf == null) {
            buf = byteBufAllocator.ioBuffer(kcp.getMtu());
        } else if (buf.readableBytes() + len > kcp.getMtu()) {
            output(buf);
            buf = byteBufAllocator.ioBuffer(kcp.getMtu());
        }
        return buf;
    }
    /**
     *  根据拥塞窗口，计算剩余可发送容量，将 que -> buf
     *  <p> 剩余容量：sndUna + cwnd，sndNxt只能小于这个上界
     */
    public void moveQueToBuf(int cwnd, int wnd) {
        long limit = sndUna + cwnd;
        while (sndNxt < limit && !sndQue.isEmpty()) {
            KcpSegment seg = sndQue.pop();
            seg.setConv(kcp.getConv());
            seg.setCmd(IKcp.IKCP_CMD_PUSH);
            seg.setWnd(wnd);
            seg.setTs(kcp.getCurrent());
            seg.setSn(sndNxt++);
            seg.setUna(kcp.getReceiver().getRcvNxt());

            seg.setReSendTs(kcp.getCurrent());
            seg.setRto(kcp.getRto().getRxRto());
            seg.setFastAck(0);
            seg.setXmit(0);

            sndBuf.addLast(seg);
        }
    }
    /**
     *  发送SndBuf
     */
    public ByteBuf sendSndBuf(ByteBuf buf, int wnd) {
    	if (sndBuf.isEmpty()) {return buf;}
    	//  窗口resent, mnRto
        int resent = kcp.getFastResend() >= 0 ? kcp.getFastResend() : Integer.MAX_VALUE;
        int mnRto = kcp.getRto().isNoDelay() ? 0 : kcp.getRto().getRxRto() >> 3;

        //  是否有丢包，是否有快速重试
        boolean lost = false;
        boolean change = false;

        //  遍历sndBuf，判断发送
        DoubleLinkedList.Node<KcpSegment> crt = sndBuf.first();
        while (crt != sndBuf.getTAIL()) {
            boolean needSend = false;
            KcpSegment seg = crt.val;
            crt = crt.next;
            //  如果是第一次发送
            if (seg.getXmit() == 0) {
                needSend = true;
                kcp.getMetric().maxSegXmit(seg.incrXmit());
                seg.setRto(kcp.getRto().getRxRto());
                seg.setReSendTs(kcp.getCurrent() + seg.getRto() + mnRto);
            }
            //  如果超过resend仍未收到ack（在sndBuf里还没被删掉），判定位丢包。需要重试
            else if (kcp.getCurrent() >= seg.getReSendTs()) {
                needSend = true;
                kcp.getMetric().maxSegXmit(seg.incrXmit());
                kcp.getMetric().incrXmit();
                seg.setFastAck(0);
                seg.setRto(seg.getRto() + (kcp.getRto().isNoDelay() ? kcp.getRto().getRxRto() : kcp.getRto().getRxRto() >>> 1));
                seg.setReSendTs(kcp.getCurrent() + seg.getRto());
                lost = true;
            }
            //  如果超过了快速重试次数
            //  seg.fastAck标记。input时如果收到了不连续的ack（中间有些seg的ack没收到，直接收到了后面的），则小于该ack的seg.fastAck标记会累加
            if (seg.getFastAck() >= resent) {
                if (seg.getFastAck() <= kcp.getFastLimit() || kcp.getFastLimit() <= 0) {
                    change = true;
                    needSend = true;
                    kcp.getMetric().maxSegXmit(seg.incrXmit());
                    seg.setFastAck(0);
                    seg.setReSendTs(kcp.getCurrent() + seg.getRto());
                }
            }
            //  如果进过上面的判断需要发送
            if (needSend) {
                seg.setWnd(wnd);
                seg.setTs(kcp.getCurrent());
                seg.setUna(kcp.getReceiver().getRcvNxt());
                //	头体一起写
                int need = IKcp.IKCP_OVERHEAD + seg.dataLen();
                buf = createOrOutput(buf, need);
                seg.writeHead(buf);
                if (seg.dataLen() > 0) {
                    seg.writeData(buf);
                }
//                //  先写头（有可能消息中不含体）
//                buf = createOrOutput(buf, IKcp.IKCP_OVERHEAD);
//                seg.writeHead(buf);
//                //  再写体
//                if (seg.dataLen() > 0) {
//                    buf = createOrOutput(buf, seg.dataLen());
//                    seg.writeData(buf);
//                }
                //  如果xmit >= deadLink，status置为-1
                if (seg.getXmit() >= kcp.getDeadLink()) {
                    kcp.setState(-1);
                }
            }
        }
        //  如果发生了快速重连，减小拥塞窗口阈值，适当缩小拥塞窗口
        if (change) {
            kcp.getCongestion().updateSsThresh(resent);
        }
        //  如果发生了丢包，拥塞窗口置为1
        if (lost) {
            kcp.getCongestion().resetCwnd(kcp.getCongestion().cwnd());
        }
        kcp.getCongestion().cwndMin1();
        return buf;
    }
    /**
     *  真实写入数据
     */
    public void output(ByteBuf buf) {
        if (buf.readableBytes() == 0) {
            return;
        }
        output.out(buf, kcp);
    }


    /**
     *  根据una清掉已经确认收到的数据
     */
    public void parseUna(long una) {
        for (KcpSegment seg = sndBuf.peek(); seg != null && seg.getSn() < una ; seg = sndBuf.peek()) {
            seg.recycle(true);
            sndBuf.pop();
        }
    }
    /**
     *  重置sndUna
     */
    public void shrinkBuf() {
        if (!sndBuf.isEmpty()) {
            sndUna = sndBuf.peek().getSn();
        } else {
            sndUna = sndNxt;
        }
    }


    /**
     *  确认sn号已经收到了
     */
    public void parseAck(long sn) {
        if (sn < 0 || sn < sndUna || sn >= sndNxt) { return; }
        //  从头开始确认，一旦sn < 当前遍历sn，那后面的更<
        sndBuf.parseAck(sn);
    }


    /**
     *  处理fastack
     */
    public void parseFastack(long ackSn) {
        if (ackSn < sndUna || ackSn >= sndNxt) { return; }
        //  sndBuf中 < ackSn 的 消息，fastAck标记累加一次
        sndBuf.parseFastack(ackSn);
    }


    /**
     *  检测待发送列表中的每一个seg，其resendts是否小于current。若小于，则该返回该current
     */
    public int checkSndBufReSendTs(int current, int tsFlush) {
        int tmFlush = tsFlush - current;
        int tmPacket = Integer.MAX_VALUE;

        DoubleLinkedList.Node<KcpSegment> crt = sndBuf.first();
        while (crt != sndBuf.getTAIL()) {
            KcpSegment seg = crt.val;
            crt = crt.next;
            int diff = seg.getReSendTs() - current;
            if (diff <= 0) {
                return current;
            }
            if (diff < tmPacket) {
                tmPacket = diff;
            }
        }
        int minimal = tmPacket < tmFlush ? tmPacket : tmFlush;
        if (minimal >= kcp.getInterval()) {
            minimal = kcp.getInterval();
        }
        return current + minimal;
    }


    /**
     * buf中的数据 -> sndQue
     * @return -2: 切出的frg数量已经超过对方的接受窗口，根据窗口逻辑已无法还原出一个完整消息<br/>
     *         >0: 实际写入的字节数
     */
    public int sendToSndQue(ByteBuf buf) {
        int len = buf.readableBytes();
        if (len == 0) { return 0; }

        //  流模式，把数据量尽量塞满最后一个Segment（会带来粘包问题，需要接收端自行解码）
        if (kcp.isStream()) {
            DoubleLinkedList.Node<KcpSegment> last = sndQue.last();
            if (last != sndQue.getHEAD()) {
                KcpSegment seg = last.val;
                int lastLen = seg.dataLen();
                //  如果最后一个seg的可读取 < mss，说明还有容量可以塞
                if (lastLen < kcp.getMss()) {
                    int capacity = kcp.getMss() - lastLen;
                    int need = Math.min(len, capacity);
                    if (seg.getData().maxWritableBytes() < need) {
                        ByteBuf newData = byteBufAllocator.ioBuffer(lastLen + need);
                        newData.writeBytes(seg.getData());
                        seg.getData().release();
                        seg.setData(newData);
                    }
                    //  将need的数据写入最后一个seg，塞满一个mss
                    seg.getData().writeBytes(buf, need);
                    len = buf.readableBytes();
                    if (len == 0) { return need; }
                }
            }
        }

        //  总共要切的frg数
        int count = len <= kcp.getMss() ? 1 : (len + kcp.getMss() -1) / kcp.getMss();
        if (!kcp.isStream() && count >= RpcKcpConfigInfo.INSTANCE.getiKcpWndRcv()) { return -2; }
        count = Math.max(count, 1);

        int send = 0;
        //  每个分块顺序写入，但每个分块的frg倒排（接收端顺序读到0，就知道是一个完整消息的结束）
        for (int i = 0 ; i < count ; i++) {
            int size = Math.min(len, kcp.getMss());

            KcpSegment seg = KcpSegment.alloc(buf.readRetainedSlice(size));
            seg.setFrg((short) (kcp.isStream() ? 0 : count - i - 1));
            sndQue.addLast(seg);

            len = buf.readableBytes();
            send += size;
        }
        return send;
    }


    /**
     *  释放
     */
    public void release() {
        while (!sndBuf.isEmpty()) {
            sndBuf.pop().recycle(true);
        }
        while (!sndQue.isEmpty()) {
            sndQue.pop().recycle(true);
        }
    }


    public long getSndUna() {
        return sndUna;
    }

    public void setSndUna(long sndUna) {
        this.sndUna = sndUna;
    }

    public long getSndNxt() {
        return sndNxt;
    }

    public void setSndNxt(long sndNxt) {
        this.sndNxt = sndNxt;
    }

    public int getSndWnd() {
        return sndWnd;
    }

    public void setSndWnd(int sndWnd) {
        this.sndWnd = sndWnd;
    }

    public KcpSenderLinkedList getSndBuf() {
        return sndBuf;
    }

    public void setSndBuf(KcpSenderLinkedList sndBuf) {
        this.sndBuf = sndBuf;
    }

    public KcpSenderLinkedList getSndQue() {
        return sndQue;
    }

    public void setSndQue(KcpSenderLinkedList sndQue) {
        this.sndQue = sndQue;
    }

    public ByteBufAllocator getByteBufAllocator() {
        return byteBufAllocator;
    }

    public void setByteBufAllocator(ByteBufAllocator byteBufAllocator) {
        this.byteBufAllocator = byteBufAllocator;
    }

	public IKcpOutput getOutput() {
		return output;
	}

	public void setOutput(IKcpOutput output) {
		this.output = output;
	}
}
